package com.example.capsule_netty.impl;

import com.example.capsule_netty.bean.PositionBean;
import com.example.capsule_netty.bean.UserInfoBean;
import com.example.capsule_netty.configuration.KafkaConstant;
import com.example.capsule_netty.mapper.UserInfoMapper;
import com.example.capsule_netty.service.UserInfoService;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.geo.*;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhuzhiming
 * @Date: 2022/5/18 12:46
 */
@Service
@Slf4j
public class PositionServiceImpl {

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    RedisTemplate<String, String> redisTemplate;

    @Resource
    UserInfoMapper userInfoMapper;

    public void uploadPosition(PositionBean positionBean) {

        kafkaTemplate.send(KafkaConstant.POSITION_TOPIC, new Gson().toJson(positionBean)).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("send upload position message fail");
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {

            }
        });

    }

    public String getNearPeople(PositionBean positionBean) {

        String key = positionBean.getField() + "-Position";
        String userId = positionBean.getUserId();
        List<Point> position = redisTemplate.opsForGeo().position(key, userId);
        Circle circle = new Circle(position.get(0), new Distance(500, Metrics.MILES));
        RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs()
                .includeDistance()
                .includeCoordinates()
                .sortAscending()
                .limit(500);
        GeoResults<RedisGeoCommands.GeoLocation<String>> results = redisTemplate.opsForGeo()
                .radius(key, circle, args);//params: key, Circle, GeoRadiusCommandArgs
        List<PositionBean> positionBeanList = new ArrayList<>();
        PositionBean bean = new PositionBean();
        for (GeoResult<RedisGeoCommands.GeoLocation<String>> result : results) {
            if (result.getContent().getName().equals(userId)) {
                continue;
            }
            bean.setUserId(result.getContent().getName());
            bean.setDistance(result.getDistance().toString());
            positionBeanList.add(bean);
        }

        for (PositionBean posBean : positionBeanList) {
            String userJson = redisTemplate.opsForValue().get(posBean.getUserId());
            String uploadTime = redisTemplate.opsForValue().get(posBean.getUserId() + "-uploadPositionTime");
            if (userJson != null) {
                UserInfoBean userInfoBean = new Gson().fromJson(userJson, UserInfoBean.class);
                posBean.setUserName(userInfoBean.getNickname());
                posBean.setSex(userInfoBean.getSex());
            } else {
                UserInfoBean userInfo = userInfoMapper.selectById(posBean.getUserId());
                posBean.setUserName(userInfo.getNickname());
                posBean.setSex(userInfo.getSex());
                redisTemplate.opsForValue().set(posBean.getUserId(), new Gson().toJson(userInfo),10, TimeUnit.MINUTES);
            }
            posBean.setUploadTime(uploadTime == null ? "10分钟之前" : uploadTime);
        }
        return new Gson().toJson(positionBeanList);
    }

    public String getDistance(PositionBean positionBean) {
        String key = positionBean.getField().toLowerCase() + "-Position";
        String userId = positionBean.getUserId();
        String otherId = positionBean.getOtherId();
        Distance distance = redisTemplate.opsForGeo().distance(key, userId, otherId, RedisGeoCommands.DistanceUnit.MILES);
        if (distance != null) {
            positionBean.setDistance(distance.toString());
        }
        if (positionBean.getUserName() == null || positionBean.getUserName().equals("")) {
            UserInfoBean userInfoBean = userInfoMapper.selectById(otherId);
            if(userInfoBean != null){
                positionBean.setUserName(userInfoBean.getNickname());
                positionBean.setSex(userInfoBean.getSex());
            }
        }
        String uploadTime = redisTemplate.opsForValue().get(otherId + "-uploadPositionTime");
        if (uploadTime == null) {
            uploadTime = "10分钟之前";
        }
        positionBean.setUploadTime(uploadTime);
        positionBean.setUserId(otherId);
        return new Gson().toJson(positionBean);

    }
}
